package com.awifi.demo.consumer;

import com.awifi.demo.conf.KafkaConsumerConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.Collections;

/**
 * @description: ToDo
 * @author: marin
 * @time: 2020/8/7 15:52
 */
@Slf4j
@Component
public class KafkaConsumerExample {


    @Autowired
    private KafkaConsumerConfig kafkaConsumerConfig;

    @Value("${kafka.consumer.topic}")
    private String topic;


    public void run() {
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(kafkaConsumerConfig.getCommonProperties());
        // 订阅test1 topic
        consumer.subscribe(Collections.singletonList(topic));

        while(true) {
            //  从服务器开始拉取数据
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
            records.forEach(record -> {
                log.error("topic:{},key:{},value:{}",record.topic(),record.key(),record.value());
            });
        }
    }
}
